package com.pony.test;

import java.io.IOException;

import net.paoding.analysis.analyzer.PaodingAnalyzer;

import org.apache.hadoop.io.IntWritable;
import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.ObjectWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapred.MapReduceBase;
import org.apache.hadoop.mapred.Mapper;
import org.apache.hadoop.mapred.OutputCollector;
import org.apache.hadoop.mapred.Reporter;
import org.apache.lucene.analysis.Analyzer;
import org.apache.lucene.document.Document;
import org.apache.lucene.document.Field;
import org.jsoup.Jsoup;


public class IndexMapper extends MapReduceBase implements
		Mapper<LongWritable, Text, IntWritable, IntermediateForm> {

	@Override
	public void map(LongWritable inputKey, Text inputValue,
			OutputCollector<IntWritable, IntermediateForm> outputCollector, Reporter reporter)
			throws IOException {
		
		String url = inputValue.toString();
		String content = getContent(url);
		
		Analyzer analyzer = new PaodingAnalyzer();
		
		Field contentField = new Field("content",content,Field.Store.NO,
                Field.Index.ANALYZED, Field.TermVector.WITH_POSITIONS_OFFSETS);
		Document doc = new Document();
		doc.add(contentField);
		
		IntermediateForm form = new IntermediateForm();
		form.process(doc,analyzer);
		form.closeWriter();
		IntWritable outKey = new IntWritable(1);
		outputCollector.collect(outKey, form);
		
	}
	
	private String getContent(String uri) throws IOException{
		org.jsoup.nodes.Document doc = Jsoup.connect(uri).get();
		return doc.outerHtml();
	}

}
